/**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2016-2026 hmc,Inc.All Rights Reserved.
 */
package com.sys.kafka.config;

import com.sys.midware.config.core.ConfigContainer;
import com.sys.midware.config.core.ConfigKey;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * 
 * <b>ClassName：</b> ConfigMongoUtil <br/>
 * <b>Description：</b> 获取配置信息工具类 <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月19日 下午2:55:02 <br/>
 * <b>@version: </b> <br/>
 */
public class KafkaConfigUtil {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConfigUtil.class);

    /**
     * ConfigContainer instance
     */
    private static ConfigContainer cfg = ConfigContainer.getInstance(ConfigKey.ConfigFile.CONFIG_FILE_ROOT);

    private static boolean inited;
    public static void initConfig() {
        if(inited){
            return;
        }
        logger.info("开始初始化kakfa配置...");
        KafkaConfig config = new KafkaConfig();
        
        config.setBootstrapServers(cfg.getParamString(KafkaConfigKey.KEY_BOOTSTRAP_SERVERS));
        
        KafkaConfig4Producer config4producer = new KafkaConfig4Producer();
        config4producer.setAcks(cfg.getParamString(KafkaConfigKey.KEY_PRODUCER_ACKS));
        config4producer.setBatch_size(cfg.getParamInteger(KafkaConfigKey.KEY_PRODUCER_BATCH_SIZE));
        config4producer.setBlock_on_buffer_full(cfg.getParamBoolean(KafkaConfigKey.KEY_PRODUCER_BLOCK_ON_BUFFER_FULL));
        config4producer.setBuffer_memory(cfg.getParamLong(KafkaConfigKey.KEY_PRODUCER_BUFFER_MEMORY));
        config4producer.setKey_serializer(cfg.getParamString(KafkaConfigKey.KEY_PRODUCER_KEY_SERIALIZER));
        config4producer.setLinger_ms(cfg.getParamLong(KafkaConfigKey.KEY_PRODUCER_LINGER_MS));
        config4producer.setReconnect_backoff_ms(cfg.getParamLong(KafkaConfigKey.KEY_PRODUCER_RECONNECT_BACKOFF_MS));
        config4producer.setRetries(cfg.getParamInteger(KafkaConfigKey.KEY_PRODUCER_RETRIES));
        config4producer.setTimeout_ms(cfg.getParamLong(KafkaConfigKey.KEY_PRODUCER_TIMEOUT_MS));
        config4producer.setValue_serializer(cfg.getParamString(KafkaConfigKey.KEY_PRODUCER_VALUE_SERIALIZER));
        
        
        config.setConfig4producer(config4producer);
        KafkaConfig4Consumer config4Consumer = new KafkaConfig4Consumer();
        config4Consumer.setAuto_commit_interval_ms(cfg.getParamLong(KafkaConfigKey.KEY_CONSUMER_AUTO_COMMIT_INTERVAL_MS));
        config4Consumer.setEnable_auto_commit(cfg.getParamBoolean(KafkaConfigKey.KEY_CONSUMER_ENABLE_AUTO_COMMIT));
        config4Consumer.setGroup_id(cfg.getParamString(KafkaConfigKey.KEY_CONSUMER_GROUP_ID));
        config4Consumer.setKey_serializer(cfg.getParamString(KafkaConfigKey.KEY_CONSUMER_KEY_SERIALIZER));
        config4Consumer.setSession_timeout_ms(cfg.getParamInteger(KafkaConfigKey.KEY_CONSUMER_SESSION_TIMEOUT_MS));
        config4Consumer.setValue_serializer(cfg.getParamString(KafkaConfigKey.KEY_CONSUMER_VALUE_SERIALIZER));
        config4Consumer.setAuto_offset_reset(cfg.getParamString(KafkaConfigKey.KEY_CONSUMER_AUTO_OFFSET_RESET));
        
        config.setConfig4consumer(config4Consumer);
        KafkaConfigHolder.put(config);
        logger.info("kakfa配置初始化完成，config:" + config);
        inited= true;
    }

}
